package com.learn.sunday.es.flink;

import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Date;
import java.util.Random;
import java.util.concurrent.atomic.AtomicLong;


public class EventTimeDemo {

    private static final Logger log = LoggerFactory.getLogger(EventTimeDemo.class);
    private static AtomicLong seq = new AtomicLong(0);

    public static void main(String[] args) throws Exception {
        // 创建执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        // 设置事件时间特性
        env.setStreamTimeCharacteristic(org.apache.flink.streaming.api.TimeCharacteristic.EventTime);

        // 创建模拟数据源
        DataStream<JSONObject> stream = env.addSource(new EventGenerator());

        // 提取事件时间戳和水印
        WatermarkStrategy<JSONObject> watermarkStrategy = WatermarkStrategy
                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner((event, timestamp) -> event.getLong("time"));
        OutputTag<JSONObject> late = new OutputTag<JSONObject>("late");
        // 添加水印和事件时间分配器
        DataStream<JSONObject> eventStream = stream.assignTimestampsAndWatermarks(watermarkStrategy);
        SingleOutputStreamOperator<JSONObject> name = eventStream.keyBy(event -> event.getString("name"))
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .sideOutputLateData(late)
                .process(new ProcessWindowFunction<JSONObject, JSONObject, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<JSONObject, JSONObject, String, TimeWindow>.Context context, Iterable<JSONObject> iterable, Collector<JSONObject> collector) throws Exception {
                        long start = context.window().getStart();
                        long end = context.window().getEnd();
                        long l = context.currentWatermark();
                        log.info("start: {}, end: {},Watermark:{}",
                                DateUtil.format(new Date(start), DatePattern.NORM_DATETIME_FORMAT),
                                DateUtil.format(new Date(end), DatePattern.NORM_DATETIME_FORMAT),
                                DateUtil.format(new Date(l), DatePattern.NORM_DATETIME_FORMAT));

                        iterable.forEach(collector::collect);
                    }
                });


        // 打印结果
        name.print();
        name.getSideOutput(late).print();

        // 执行程序
        env.execute("Event Time Demo with Simulated Source");
    }

    // 模拟数据源
    private static class EventGenerator implements SourceFunction<JSONObject> {
        private volatile boolean running = true;
        private Random random = new Random();

        @Override
        public void run(SourceContext<JSONObject> ctx) throws Exception {
            while (running) {
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("id", seq.getAndAdd(1L));
                jsonObject.put("name", "channelCode" + random.nextInt(10));
                if (random.nextInt(5) % 2 == 0) {
                    jsonObject.put("time", System.currentTimeMillis() + random.nextInt(50000));
                } else {
                    jsonObject.put("time", System.currentTimeMillis() - random.nextInt(50000));
                }
                ctx.collect(jsonObject);
                Thread.sleep(random.nextInt(10000)); // 随机延迟
            }
        }

        @Override
        public void cancel() {
            running = false;
        }
    }
}



